package articleEvents

import (
	"context"
	"github.com/IBM/sarama"
	dao "jk-time/webook/internal/dao/article"
	"jk-time/webook/internal/domain"
	"jk-time/webook/internal/repository"
	"jk-time/webook/pkg/canalx"
	"jk-time/webook/pkg/logger"
	"jk-time/webook/pkg/saramax"
	"time"
)

type MySQLBinlogComsumer struct {
	client sarama.Client
	repo   *repository.CacheArticleRepository
	l      logger.Logger
}

func (k *MySQLBinlogComsumer) Start() error {
	cg, err := sarama.NewConsumerGroupFromClient("mysql_binlog", k.client)
	if err != nil {
		return err
	}
	go func() {
		err := cg.Consume(context.Background(),
			[]string{"example"},
			saramax.NewHandler[canalx.Message[dao.PublishedArticle]]("mysql_binlog", k.Consume))
		if err != nil {
			k.l.Error("退出了消费循环,异常", logger.Error(err))
		}
	}()
	return nil
}

func (k *MySQLBinlogComsumer) Consume(msg *sarama.ConsumerMessage, cmsg canalx.Message[dao.PublishedArticle]) error {
	if cmsg.Table != "published_articles" {
		return nil
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	for _, data := range cmsg.Data {
		var err error
		switch data.Status {
		case domain.ArticleStatusPublished.ToUint8():
			err = k.repo.Cache().SetPub(ctx, k.repo.ToDomain(data))
		case domain.ArticleStatusPrivate.ToUint8():
			err = k.repo.Cache().DelPub(ctx, data.Id)
		}
		if err != nil {
			return err
		}
	}
	return nil
}
